package net.sina.realtime.traffic.job;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.sina.realtime.traffic.utils.DateTimeUtil;
import net.sina.realtime.traffic.utils.JdbcUtil;
import net.sina.realtime.traffic.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
public class TransportationWindowDwsJob {
    public static void main(String[] args) throws Exception {
        //1.执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.数据源-source
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "traffic-log");
//        System.out.println(kafkaDataStream);
        kafkaDataStream.print();
        //3.数据转换-transformation
        handle(kafkaDataStream);
        //4.触发执行-execute
        env.execute("TransportationWindowDwsJob");
    }

    private static void handle(DataStream<String> kafkaDataStream) {
        // 设置水位线
        SingleOutputStreamOperator<String> watermarks = kafkaDataStream.
                assignTimestampsAndWatermarks(WatermarkStrategy.
                        <String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        return JSON.parseObject(element).getLong("timestamp");
                    }
                })
        );

        // 分组
        KeyedStream<String, String> roadId = watermarks.keyBy(
                json -> JSON.parseObject(json).getString("road_id")
        );

        // 窗口
        WindowedStream<String, String, TimeWindow> windowStream = roadId.window(
                TumblingEventTimeWindows.of(Time.seconds(10))
        );


        SingleOutputStreamOperator<String> process = windowStream.process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                // 获取道路ID
                ArrayList<String> list = ListUtil.toList(elements);
                String roadId = JSONObject.parseObject(list.get(0)).getString("road_id");
                // 获取窗口的开始时间、结束时间
                String window_start_time = DateTimeUtil.convertLongToString(context.window().getStart(), DateTimeUtil.DATE_TIME_FORMAT);
                String window_end_time = DateTimeUtil.convertLongToString(context.window().getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

                // 遍历窗口内的所有事件，统计车辆数和总速度
                long severeCount = 0;
                double totalSpeed = 0.0;
                for (String element : elements) {
                    severeCount++;
                    totalSpeed += JSON.parseObject(element).getDouble("speed");
                }

                long ts = System.currentTimeMillis();
                String output =
                        window_start_time + "," +
                        window_end_time + "," +
                        roadId + ","  +
                        severeCount+ "," +
                        totalSpeed + "," +
                        ts;
                out.collect(output);
//                System.out.println(output);
            }

        });


        JdbcUtil.sinkToClickhouseUpsert(process, "insert into traffic_log_report.dwd_congestion (" +
                " window_start_time," +
                " window_end_time," +
                " road_id," +
                " severeCount," +
                " avgspeed,\n" +
                " ts)\n" +
                "values (?,?,?,?,?,?)");

    }
}
